#include "base_shared_ring_buffer.h"
#include "base_log.h"

#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <cstring>
#include <cerrno>
#include <thread>
#include <chrono>
#include <algorithm>
#include <ctime>

namespace El {
namespace Base {

// 魔数，用于验证共享内存格式
constexpr uint32_t RING_BUFFER_MAGIC = 0x52494E47; // "RING" in ASCII

SharedRingBuffer::SharedRingBuffer()
    : shmFd_(-1),
      shmAddr_(nullptr),
      header_(nullptr),
      buffer_(nullptr),
      bufferSize_(0),
      isWriter_(false),
      myReaderId_(INVALID_READER_ID),
      readerSemaphore_(nullptr)
{
}

SharedRingBuffer::~SharedRingBuffer()
{
    Close();
}

std::shared_ptr<SharedRingBuffer> SharedRingBuffer::Create(const std::string &name, size_t size, bool isWriter)
{
    std::shared_ptr<SharedRingBuffer> ringBuffer(new SharedRingBuffer());
    if (!ringBuffer->Init(name, size, isWriter)) {
        return nullptr;
    }
    return ringBuffer;
}

bool SharedRingBuffer::Init(const std::string &name, size_t size, bool isWriter)
{
    // 保存共享内存名称和基础信号量名称
    shmName_ = name;

    // 处理名称，确保兼容性
    std::string baseName;
    size_t lastSlash = name.find_last_of('/');
    if (lastSlash != std::string::npos) {
        baseName = name.substr(lastSlash + 1);
    } else {
        baseName = name;
    }
    baseSemName_ = "/sem_rb_" + baseName;

    isWriter_ = isWriter;

    // 计算实际需要的共享内存大小（头部 + 数据区）
    size_t totalSize = sizeof(RingBufferHeader) + size;

    // 打开或创建共享内存
    int flags = O_RDWR;
    if (isWriter) {
        flags |= O_CREAT;
    }

    // 确保名称以/开头，这是POSIX共享内存的要求
    std::string fullShmName = name;
    if (!name.empty() && name[0] != '/') {
        fullShmName = "/" + name;
    }

    // 打开共享内存对象
    shmFd_ = shm_open(fullShmName.c_str(), flags, 0666);
    if (shmFd_ == -1) {
        if (isWriter && errno == ENOENT) {
            // 如果是写入者且文件不存在，尝试清理并重试
            shm_unlink(fullShmName.c_str());
            shmFd_ = shm_open(fullShmName.c_str(), flags, 0666);
            if (shmFd_ == -1) {
                return false;
            }
        } else {
            return false;
        }
    }

    // 如果是写入者，需要设置共享内存大小
    if (isWriter) {
        if (ftruncate(shmFd_, totalSize) == -1) {
            close(shmFd_);
            shmFd_ = -1;
            shm_unlink(fullShmName.c_str());
            return false;
        }
    }

    // 映射共享内存到进程地址空间
    shmAddr_ = mmap(nullptr, totalSize, PROT_READ | PROT_WRITE, MAP_SHARED, shmFd_, 0);
    if (shmAddr_ == MAP_FAILED) {
        close(shmFd_);
        shmFd_ = -1;
        if (isWriter) {
            shm_unlink(fullShmName.c_str());
        }
        return false;
    }

    // 设置头部和数据区指针
    header_ = static_cast<RingBufferHeader *>(shmAddr_);
    buffer_ = static_cast<uint8_t *>(shmAddr_) + sizeof(RingBufferHeader);
    bufferSize_ = size;

    // 如果是写入者，初始化头部
    if (isWriter) {
        header_->magic = RING_BUFFER_MAGIC;
        header_->bufferSize = size;
        header_->isWriting = false;
        header_->writeCount = 0;
        header_->writePos = 0;
        header_->minReadPos = 0;
        header_->readerCount = 0;
        header_->nextReaderId = 0;
        header_->activeReaders = 0; // 确保初始化activeReaders

        // 初始化心跳时间戳
        header_->writerHeartbeat.store(static_cast<uint64_t>(time(nullptr)), std::memory_order_release);

        // 初始化所有读者位置
        for (uint32_t i = 0; i < 32; i++) {
            header_->readerPositions[i].store(0, std::memory_order_relaxed);
        }
    } else {
        // 验证魔数
        if (header_->magic != RING_BUFFER_MAGIC) {
            return false;
        }

        // 验证缓冲区大小
        if (header_->bufferSize != size) {
            return false;
        }

        // 如果是读者，需要注册
        myReaderId_ = RegisterReader();
        if (myReaderId_ == INVALID_READER_ID) {
            return false;
        }

        // 为自己创建信号量
        if (!CreateOrOpenReaderSemaphore(myReaderId_)) {
            UnregisterReader(myReaderId_);
            return false;
        }
    }

    return true;
}

void SharedRingBuffer::Close()
{
    if (shmAddr_ != nullptr) {
        // 如果是读者，先注销
        if (!isWriter_ && myReaderId_ != INVALID_READER_ID) {
            UnregisterReader(myReaderId_);
            CloseReaderSemaphore(myReaderId_);
        }

        // 获取总大小
        size_t totalSize = sizeof(RingBufferHeader) + bufferSize_;

        // 解除映射
        munmap(shmAddr_, totalSize);
        shmAddr_ = nullptr;
        header_ = nullptr;
        buffer_ = nullptr;

        // 关闭文件描述符
        if (shmFd_ != -1) {
            close(shmFd_);
            shmFd_ = -1;
        }

        // 如果是写入者，删除共享内存
        if (isWriter_) {
            std::string fullShmName = shmName_;
            if (!shmName_.empty() && shmName_[0] != '/') {
                fullShmName = "/" + shmName_;
            }
            shm_unlink(fullShmName.c_str());
        }
    }
}

std::string SharedRingBuffer::GenerateReaderSemName(uint32_t readerId) const
{
    return baseSemName_ + "_r" + std::to_string(readerId);
}

bool SharedRingBuffer::CreateOrOpenReaderSemaphore(uint32_t readerId)
{
    if (readerId >= 32) {
        return false;
    }

    std::string semName = GenerateReaderSemName(readerId);

    // 打开或创建信号量
    sem_t *sem = sem_open(semName.c_str(), O_CREAT, 0666, 0);
    if (sem == SEM_FAILED) {
        return false;
    }

    readerSemaphore_ = sem;
    return true;
}

void SharedRingBuffer::CloseReaderSemaphore(uint32_t readerId)
{
    if (readerSemaphore_ != nullptr) {
        sem_close(readerSemaphore_);
        readerSemaphore_ = nullptr;
    }

    // 如果是写入者，还需要删除信号量
    if (isWriter_) {
        std::string semName = GenerateReaderSemName(readerId);
        sem_unlink(semName.c_str());
    }
}

void SharedRingBuffer::NotifyReader(uint32_t readerId)
{
    if (readerId >= 32) {
        return;
    }

    std::string semName = GenerateReaderSemName(readerId);
    sem_t *sem = sem_open(semName.c_str(), 0);
    if (sem != SEM_FAILED) {
        sem_post(sem);
        sem_close(sem);
    }
}

void SharedRingBuffer::NotifyAllReaders()
{
    uint32_t nextReaderId = header_->nextReaderId.load(std::memory_order_acquire);
    for (uint32_t i = 0; i < nextReaderId; i++) {
        NotifyReader(i);
    }
}

uint32_t SharedRingBuffer::RegisterReader()
{
    if (shmAddr_ == nullptr) {
        return INVALID_READER_ID;
    }

    // 分配读者ID
    uint32_t readerId;
    if (isWriter_) {
        readerId = header_->nextReaderId.fetch_add(1, std::memory_order_relaxed);
    } else {
        readerId = header_->nextReaderId.fetch_add(1, std::memory_order_acq_rel);
    }

    // 检查是否超出最大读者数量
    if (readerId >= 32) {
        return INVALID_READER_ID;
    }

    // 增加读者计数
    header_->readerCount.fetch_add(1, std::memory_order_relaxed);

    // 设置活跃读者位掩码
    uint32_t currentMask = header_->activeReaders.load(std::memory_order_relaxed);
    header_->activeReaders.store(currentMask | (1U << readerId), std::memory_order_release);

    // 初始化读者位置为当前写位置
    size_t currentWritePos = header_->writePos.load(std::memory_order_acquire);
    header_->readerPositions[readerId].store(currentWritePos, std::memory_order_release);

    return readerId;
}

void SharedRingBuffer::UnregisterReader(uint32_t readerId)
{
    if (shmAddr_ == nullptr || readerId >= 32) {
        return;
    }

    // 减少读者计数
    uint32_t previousCount = header_->readerCount.fetch_sub(1, std::memory_order_relaxed);

    // 清除活跃读者位掩码
    uint32_t currentMask = header_->activeReaders.load(std::memory_order_relaxed);
    header_->activeReaders.store(currentMask & ~(1U << readerId), std::memory_order_release);

    if (previousCount <= 1) {
        // 没有读者了，重置最小读位置
        header_->minReadPos.store(header_->writePos.load(std::memory_order_relaxed), std::memory_order_relaxed);
    } else {
        // 更新最小读位置
        if (isWriter_) {
            UpdateMinReadPosition();
        }
    }
}

void SharedRingBuffer::UpdateMinReadPosition()
{
    if (shmAddr_ == nullptr || !isWriter_) {
        return;
    }

    uint32_t readerCount = header_->readerCount.load(std::memory_order_relaxed);
    if (readerCount == 0) {
        return;
    }

    // 查找所有活跃读者中的最小读位置
    size_t minPos = SIZE_MAX;
    for (uint32_t i = 0; i < header_->nextReaderId.load(std::memory_order_relaxed); i++) {
        size_t pos = header_->readerPositions[i].load(std::memory_order_relaxed);
        minPos = std::min(minPos, pos);
    }

    if (minPos != SIZE_MAX) {
        header_->minReadPos.store(minPos, std::memory_order_release);
    }
}

size_t SharedRingBuffer::GetReaderPosition(uint32_t readerId) const
{
    if (shmAddr_ == nullptr || readerId >= 32) {
        return 0;
    }

    // 直接从共享内存读取
    return header_->readerPositions[readerId].load(std::memory_order_acquire);
}

void SharedRingBuffer::SetReaderPosition(uint32_t readerId, size_t position)
{
    if (shmAddr_ == nullptr || readerId >= 32) {
        return;
    }

    // 直接更新共享内存
    header_->readerPositions[readerId].store(position, std::memory_order_release);

    // 如果是写入者，更新最小读位置
    if (isWriter_) {
        UpdateMinReadPosition();
    }
}

ssize_t SharedRingBuffer::Write(const void *data, size_t size)
{
    if (shmAddr_ == nullptr || data == nullptr || size == 0) {
        return -1;
    }

    // 只有写入者可以写入数据
    if (!isWriter_) {
        return -1;
    }

    // 设置写入标志
    header_->isWriting.store(true, std::memory_order_release);

    // 计算可写空间大小
    size_t minReadPos = header_->minReadPos.load(std::memory_order_acquire);
    size_t writePos = header_->writePos.load(std::memory_order_relaxed);
    size_t freeSize;

    if (minReadPos <= writePos) {
        // 读位置在写位置之前或相等
        freeSize = bufferSize_ - writePos + minReadPos;
    } else {
        // 读位置在写位置之后
        freeSize = minReadPos - writePos;
    }

    // 预留一个字节，避免读写位置重叠时无法区分缓冲区空和满的状态
    if (freeSize <= 1) {
        header_->isWriting.store(false, std::memory_order_release);
        return 0; // 缓冲区已满
    }

    // 调整写入大小
    size_t writeSize = (size < freeSize - 1) ? size : (freeSize - 1);

    // 写入数据
    const uint8_t *src = static_cast<const uint8_t *>(data);
    if (minReadPos <= writePos) {
        // 读位置在写位置之前或相等，可能需要分两段写入
        size_t firstPart = bufferSize_ - writePos;
        if (writeSize <= firstPart) {
            // 一次性写入
            memcpy(buffer_ + writePos, src, writeSize);
        } else {
            // 分两段写入
            memcpy(buffer_ + writePos, src, firstPart);
            memcpy(buffer_, src + firstPart, writeSize - firstPart);
        }
    } else {
        // 读位置在写位置之后，可以一次性写入
        memcpy(buffer_ + writePos, src, writeSize);
    }

    // 增加写入计数
    header_->writeCount.fetch_add(1, std::memory_order_release);

    // 更新写位置
    header_->writePos.store((writePos + writeSize) % bufferSize_, std::memory_order_release);

    // 释放写入标志
    header_->isWriting.store(false, std::memory_order_release);

    // 写入完成后通知所有读者
    NotifyAllReaders();

    return writeSize;
}

ssize_t SharedRingBuffer::Read(void *data, size_t size, uint32_t readerId)
{
    if (shmAddr_ == nullptr || data == nullptr || size == 0) {
        return -1;
    }

    // 如果未指定读者ID，使用当前进程的读者ID
    if (readerId == DEFAULT_READER_ID && !isWriter_) {
        readerId = myReaderId_;
    }

    if (readerId >= 32) {
        return -1;
    }

    // 读取当前读者位置和写入位置
    size_t readPos = GetReaderPosition(readerId);
    size_t writePos = header_->writePos.load(std::memory_order_seq_cst);

    // 计算可读数据大小
    size_t dataSize;
    if (readPos <= writePos) {
        // 读位置在写位置之前或相等
        dataSize = writePos - readPos;
    } else {
        // 读位置在写位置之后
        dataSize = bufferSize_ - readPos + writePos;
    }

    if (dataSize == 0) {
        return 0; // 缓冲区为空
    }

    // 调整读取大小
    size_t readSize = (size < dataSize) ? size : dataSize;

    // 读取数据
    uint8_t *dst = static_cast<uint8_t *>(data);
    if (readPos <= writePos) {
        // 读位置在写位置之前或相等，可以一次性读取
        memcpy(dst, buffer_ + readPos, readSize);
    } else {
        // 读位置在写位置之后，可能需要分两段读取
        size_t firstPart = bufferSize_ - readPos;
        if (readSize <= firstPart) {
            // 一次性读取
            memcpy(dst, buffer_ + readPos, readSize);
        } else {
            // 分两段读取
            memcpy(dst, buffer_ + readPos, firstPart);
            memcpy(dst + firstPart, buffer_, readSize - firstPart);
        }
    }

    // 更新读者位置
    SetReaderPosition(readerId, (readPos + readSize) % bufferSize_);

    return readSize;
}

size_t SharedRingBuffer::GetReadableSize(uint32_t readerId) const
{
    if (shmAddr_ == nullptr) {
        return 0;
    }

    // 如果未指定读者ID，使用当前进程的读者ID
    if (readerId == DEFAULT_READER_ID && !isWriter_) {
        readerId = myReaderId_;
    }

    if (readerId >= 32) {
        return 0;
    }

    size_t readPos = GetReaderPosition(readerId);
    size_t writePos = header_->writePos.load(std::memory_order_seq_cst);

    if (readPos <= writePos) {
        return writePos - readPos;
    } else {
        return bufferSize_ - readPos + writePos;
    }
}

size_t SharedRingBuffer::GetWritableSize() const
{
    if (shmAddr_ == nullptr || !isWriter_) {
        return 0;
    }

    size_t minReadPos = header_->minReadPos.load(std::memory_order_acquire);
    size_t writePos = header_->writePos.load(std::memory_order_relaxed);

    // 预留一个字节，避免读写位置重叠时无法区分缓冲区空和满的状态
    if (minReadPos <= writePos) {
        return bufferSize_ - writePos + minReadPos - 1;
    } else {
        return minReadPos - writePos - 1;
    }
}

void SharedRingBuffer::Clear()
{
    if (shmAddr_ == nullptr || !isWriter_) {
        return;
    }

    // 锁定写入
    header_->isWriting.store(true, std::memory_order_seq_cst);

    // 获取当前写入位置，使用seq_cst来确保完全同步
    size_t writePos = header_->writePos.load(std::memory_order_seq_cst);

    // 清空所有读者位置（设置为当前写位置）
    uint32_t readerCount = header_->readerCount.load(std::memory_order_seq_cst);
    if (readerCount > 0) {
        for (uint32_t i = 0; i < header_->nextReaderId.load(std::memory_order_seq_cst); i++) {
            if ((header_->activeReaders.load(std::memory_order_seq_cst) & (1U << i)) != 0) {
                SetReaderPosition(i, writePos);
            }
        }
    }

    // 增加写入计数通知读者
    header_->writeCount.fetch_add(1, std::memory_order_seq_cst);

    // 释放写入锁
    header_->isWriting.store(false, std::memory_order_seq_cst);

    // 通知所有读者
    NotifyAllReaders();
}

void SharedRingBuffer::UpdateHeartbeat()
{
    if (!isWriter_ || !header_) {
        return;
    }

    // 更新心跳时间戳（使用当前时间的秒数）
    header_->writerHeartbeat.store(static_cast<uint64_t>(time(nullptr)), std::memory_order_release);
}

bool SharedRingBuffer::IsWriterActive(uint32_t timeoutSeconds)
{
    if (isWriter_ || !header_) {
        return false; // 写入者不需要检查自己是否活跃
    }

    // 获取当前时间和最后心跳时间
    uint64_t currentTime = static_cast<uint64_t>(time(nullptr));
    uint64_t lastHeartbeat = header_->writerHeartbeat.load(std::memory_order_acquire);

    // 如果未设置心跳时间（为0），则默认视为活跃
    if (lastHeartbeat == 0) {
        return true;
    }

    // 检查心跳是否在超时范围内
    return (currentTime - lastHeartbeat) <= timeoutSeconds;
}

bool SharedRingBuffer::WaitForData(uint32_t timeoutMs, uint32_t readerId)
{
    if (shmAddr_ == nullptr) {
        return false;
    }

    // 如果未指定读者ID，使用当前进程的读者ID
    if (readerId == DEFAULT_READER_ID && !isWriter_) {
        readerId = myReaderId_;
    }

    // 如果是写入者或无效读者ID，无法等待
    if (isWriter_ || readerId >= 32) {
        return false;
    }

    // 如果已经有数据可读，直接返回
    if (GetReadableSize(readerId) > 0) {
        return true;
    }

    // 记录当前写入计数
    uint64_t currentWriteCount = header_->writeCount.load(std::memory_order_acquire);

    // 再次检查是否有数据（避免读写竞态）
    if (GetReadableSize(readerId) > 0) {
        return true;
    }

    // 没有数据时，等待信号量
    if (timeoutMs == 0) {
        // 无限等待
        if (sem_wait(readerSemaphore_) != 0) {
            return false;
        }
    } else {
        // 超时等待
        struct timespec ts;
        clock_gettime(CLOCK_REALTIME, &ts);
        ts.tv_sec += timeoutMs / 1000;
        ts.tv_nsec += (timeoutMs % 1000) * 1000000;
        if (ts.tv_nsec >= 1000000000) {
            ts.tv_sec += 1;
            ts.tv_nsec -= 1000000000;
        }

        if (sem_timedwait(readerSemaphore_, &ts) != 0) {
            if (errno == ETIMEDOUT) {
                // 正常超时
                return false;
            } else {
                return false;
            }
        }
    }

    // 检查写入计数是否变化，以确认是否有新数据
    uint64_t newWriteCount = header_->writeCount.load(std::memory_order_acquire);
    return newWriteCount > currentWriteCount || GetReadableSize(readerId) > 0;
}

} // namespace Base
} // namespace El